"""
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    created by lei.chang on '24/05/2023'
    comment: 服务记录相关
"""
import re
import time
from datetime import datetime
from datetime import timedelta
from itertools import groupby
import json
from sqlalchemy import or_, and_, func, text, case
from curl_cffi import requests

from config import Config
from project.core.exception import BizError
from project.core.base import BodyParams, Select
from project.core.function import validate_phone, serializer
from project.lib.common import logger
from project.lib.constant import TIME_PERIOD_REGEX
from project.model.basic import ProcessType, Selector, Device, SmsTemplate, Resource, SmsTemplateSuccess, SmsTemplateCancel, PhoneType, TelTemplate, TelTemplateCancel
from project.model.database.flow import Flow, FlowProcess, FlowRole, FlowPage
from project.model.database.node import Node
from project.model.database.record import FlowRecord, FlowRecordDetail, FlowRecordLog
from project.model.database.cariabot import Invoice
from project.model.basic import InformType
from project.service.common.code_generator import CodeGenerator
from project.micro_service.zeus import Zeus
from project.micro_service.assichat import Assichat
from project.service.common.mqtt import MQTTService
from project.service.flow import FlowService
from project.middleware.ali_tel import AliTel
from project.core.log import Colors


class RecordService:

    @staticmethod
    async def post_third_party(session, flow_record_id, device: Device):
        """ 发送第三方
        :param session:
        :param flow_record_id:
        :param device:
        :return:
        """
        ret_data = await RecordService.get_flow_record_detail(session, flow_record_id)
        if not ret_data:
            return

        flow = await Flow.fetch_one(session, wheres=(Flow.id == ret_data['flow_id']))
        if not flow or not flow.third_party_url:
            return

        processes = await FlowProcess.fetch_all(session, wheres=(FlowProcess.flow_id == ret_data['flow_id'], FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank,))
        process_list = [{'flow_process_id': v.id, 'flow_process_name': v.name, 'flow_process_rank': v.rank} for v in processes]

        post_content = ''
        try:
            post_content = serializer(ret_data)
            # todo加工为上一版格式，
            post_content['classify_name'] = post_content.pop('flow_name', '')
            post_content['classify_process_name'] = post_content.pop('flow_process_name', '')
            flow_record_logs = post_content.pop('flow_record_logs', [])
            for flow_record_log in flow_record_logs:
                flow_record_log['classify_role_name'] = flow_record_log.pop('flow_role_name', '')
                flow_record_log['classify_role_phone'] = flow_record_log.pop('flow_role_phone', '')
                flow_record_log['classify_process_name'] = flow_record_log.pop('flow_process_name', '')
            post_content['answer_record_logs'] = flow_record_logs
            post_content['process_list'] = process_list
            requests.post(url=flow.third_party_url, headers={'deviceId': device.dev_id}, json=post_content)
            logger.info(f'第三方服务记录发送成功 url: {flow.third_party_url} post_content: {post_content}')
        except Exception as ex:
            logger.info(f'第三方服务记录发送失败{ex} url: {flow.third_party_url} post_content: {post_content}')

    @staticmethod
    async def send_sms(session, flow_record_id, sms_type='init'):
        """ 发送短信
        :param session:
        :param flow_record_id:
        :param sms_type: 短信类型 init(下单)/success(已完成)/cancel(已取消)
        :return:
        """
        if not sms_type:
            return

        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            return

        # 用来记录短信发送记录(实现幂等)
        flow_record_log = await FlowRecordLog.fetch_one(session, wheres=(FlowRecordLog.flow_record_id == flow_record.id), orders=(FlowRecordLog.id.desc()))
        if not flow_record_log or flow_record_log.is_sms_notified:
            return
        flow_record_log.is_sms_notified = True

        await session.commit()

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        if not flow:
            return

        # 下单通知
        flow_process_id = 0
        if sms_type == 'init':
            worker_ele = SmsTemplate.worker_sms.value
            user_ele = SmsTemplate.user_sms.value
            flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow.id, FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank,))
            flow_process_id = flow_process.id if flow_process else 0
        # 状态流转通知
        elif sms_type == 'transfer':
            worker_ele = SmsTemplate.worker_sms.value
            flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.id == flow_record.flow_process_id, ~FlowProcess.is_deleted))
            flow_process_id = flow_process.id if flow_process else 0
        # 完成通知
        elif sms_type == 'success':
            worker_ele = SmsTemplateSuccess.worker_sms.value
            user_ele = SmsTemplateSuccess.user_sms.value
            flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow.id, FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank.desc()))
            if not flow_process:
                flow_process_id = 0
            else:
                flow_process_id = flow_record_log.flow_process_id
        # 取消通知
        elif sms_type == 'cancel':
            worker_ele = SmsTemplateCancel.worker_sms.value
            user_ele = SmsTemplateCancel.user_sms.value
            # 查询上一个服务流程节点状态
            flow_record_log = await FlowRecordLog.fetch_one(session, wheres=(FlowRecordLog.flow_record_id == flow_record.id, FlowRecordLog.record_status != 'cancel'), orders=(FlowRecordLog.id.desc()))
            if flow_record_log and flow_record_log.flow_process_id:
                flow_process_id = flow_record_log.flow_process_id
        else:
            return

        template_params = {
            'sevice_address': flow_record.room_no or '系统',
            'sevice_name': flow.name,
            'env': Config.SMS_ENV,
            'record_id': flow_record.id,
            'reserved': 'reserved'
        }

        flow_role_ids = [v.flow_role_id for v in flow.inform_config_obj or [] if v.flow_process_id == flow_process_id and str(InformType.sms.value) in v.inform_type]
        flow_roles = await FlowRole.fetch_all(session, wheres=(FlowRole.id.in_(flow_role_ids), ~FlowRole.is_deleted))
        for flow_role in flow_roles:
            if not validate_phone(flow_role.phone):
                continue
            template_params['classify_role_id'] = flow_role.id
            await Assichat.send_sms(phone=flow_role.phone, template_params=template_params, ele=worker_ele)

        # 未设置工单状态，下单时用户短信已发送，那么完成
        if sms_type == 'success' and flow_process_id == 0:
            return

        # 状态流转时，不给用户发短信
        if sms_type == 'transfer':
            return

        # 给用户发短信
        data = await Select(selects=(FlowRecordDetail, Node),
                            joins=((Node, FlowRecordDetail.node_id == Node.id),),
                            wheres=(FlowRecordDetail.flow_record_id == flow_record.id),
                            orders=(FlowRecordDetail.id,)).fetch_all(session)

        sevice_time = flow_record.create_time.strftime('%Y-%m-%d %H:%M')
        service_content_title = flow.name
        sevice_details = ''
        phone = ''
        sevice_greet = ''
        is_send_sms = False
        for flow_record_detail, node in data:
            # 图文列表、文本列表
            if flow_record_detail.node_type in ('image_display_select', 'text_picker'):
                if flow_record_detail.count:
                    sevice_details += flow_record_detail.title + f' * {flow_record_detail.count}、'
                else:
                    sevice_details += flow_record_detail.title + '、'
            # 时间段选择
            elif flow_record_detail.node_type in ('time_period_picker', ):
                if flow_record_detail.title:
                    temp_sevice_time = f"{flow_record.create_time.strftime('%Y-%m-%d')} {flow_record_detail.title.split('~')[1]}"
                    temp_sevice_time = datetime.strptime(temp_sevice_time, '%Y-%m-%d %H:%M')
                    days = 0
                    if flow_record_detail.date == 'today':
                        days = 0
                    elif flow_record_detail.date == 'tomorrow':
                        days = 1
                    elif flow_record_detail.date == 'after_tomorrow':
                        days = 2
                    sevice_time = (temp_sevice_time + timedelta(days=days)).strftime('%Y-%m-%d %H:%M')

            # 手机号输入
            elif flow_record_detail.node_type in ('text_phone', ):
                if validate_phone(flow_record_detail.title):
                    phone = flow_record_detail.title
                    sevice_greet = node.sms_greet
                    is_send_sms = node.is_send_sms

        sevice_details = f'{service_content_title}: {sevice_details if sevice_details else flow.name}'
        if is_send_sms:
            flow_role = await FlowRole.fetch_one(session, wheres=(FlowRole.flow_id == flow.id, FlowRole.role_type == 'user', ~FlowRole.is_deleted))
            template_params['classify_role_id'] = flow_role.id if flow_role else 0
            template_params['sevice_greet'] = sevice_greet
            template_params['sevice_time'] = sevice_time
            template_params['sevice_details'] = sevice_details[0:30].rstrip('、')
            await Assichat.send_sms(phone=phone, template_params=template_params, ele=user_ele)

    @staticmethod
    async def send_tel(session, flow_record_id, tel_type='init'):
        """ 发送语音通知
        :param session:
        :param flow_record_id:
        :param tel_type:
        :return:
        """
        if not tel_type:
            return

        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            return

        # 用来记录短信发送记录(实现幂等)
        flow_record_log = await FlowRecordLog.fetch_one(session, wheres=(FlowRecordLog.flow_record_id == flow_record.id), orders=(FlowRecordLog.id.desc()))
        if not flow_record_log or flow_record_log.is_tel_notified:
            return
        flow_record_log.is_tel_notified = True

        await session.commit()

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        if not flow:
            return

        # 下单通知
        if tel_type == 'init':
            worker_ele = TelTemplate.worker_sms.value
            flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow.id, FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank,))
            flow_process_id = flow_process.id if flow_process else 0
        else:
            return

        template_params = {
            'sevice_address': flow_record.room_no or '系统',
            'sevice_name': flow.name
        }

        flow_role_ids = [v.flow_role_id for v in flow.inform_config_obj or [] if v.flow_process_id == flow_process_id and str(InformType.tel.value) in v.inform_type]
        flow_roles = await FlowRole.fetch_all(session, wheres=(FlowRole.id.in_(flow_role_ids), ~FlowRole.is_deleted))
        for flow_role in flow_roles:
            if not validate_phone(flow_role.phone):
                continue
            AliTel.send(called_number=flow_role.phone, tts_code=worker_ele.code, tts_param=str(template_params))

    @staticmethod
    async def do_flow_record_submit(session, flow_code, device: Device, selectors: list[Selector] = None, resource: Resource = None):
        """ 提交服务记录
        :param session:
        :param flow_code: 服务流程编号
        :param device: 设备信息
        :param selectors: 选择的节点信息
        :param resource: 上传
        :return:
        """
        # Step1: ==========提交服务记录==========
        flow = await Flow.fetch_one(session, wheres=(Flow.code == flow_code))
        if not flow:
            raise BizError('服务流程不存在')
        # 服务流程状态
        flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow.id, FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank, ))
        flow_process_id = flow_process.id if flow_process else 0
        flow_process_name = flow_process.name if flow_process else ''
        status = 'init' if flow_process else 'success'
        response = flow_process.digital_human_broadcast if flow_process else ''

        flow_record = await FlowRecord.create_modify(session,
                                                     business_date=datetime.now().strftime('%Y-%m-%d'),
                                                     flow_id=flow.id,
                                                     flow_code=flow.code,
                                                     organization_code=flow.organization_code,
                                                     status=status,
                                                     flow_process_id=flow_process_id,
                                                     flow_process_name=flow_process_name,
                                                     dev_id=device.dev_id,
                                                     room_no=device.room_no,
                                                     resource=resource.dict() if resource else {},
                                                     commit=False)
        if not flow_record:
            raise BizError('服务记录提交失败')
        # 生成服务记录编号
        flow_record.code = await CodeGenerator.generator_flow_record_code(flow_code)

        node_codes = [v.node_code for v in selectors or []]
        nodes = await Node.fetch_all(session, wheres=(Node.code.in_(node_codes)))
        node_map = {v.code: v for v in nodes}

        for selector in selectors or []:
            node = node_map.get(selector.node_code)
            if node.node_type in ('confirm', 'success'):
                continue
            for item in selector.items:
                if node.node_type in ('time_period_picker',):
                    if not re.match(TIME_PERIOD_REGEX, item.title):
                        continue
                elif node.node_type in ('text_phone',):
                    # 手机号(后四位)
                    if node.phone_type == str(PhoneType.last_four_bit.value):
                        if not re.match(r'^\d{4}$', item.title):
                            continue
                    # 手机号
                    elif node.phone_type == str(PhoneType.eleven_bit.value):
                        if not validate_phone(item.title):
                            continue

                await FlowRecordDetail.create_modify(session,
                                                     flow_record_id=flow_record.id,
                                                     flow_id=flow.id,
                                                     flow_code=flow.code,
                                                     node_id=node.id if node else 0,
                                                     node_code=selector.node_code,
                                                     node_type=node.node_type if node else '',
                                                     node_name=node.name if node else '',
                                                     group_title=item.group_title,
                                                     title=item.title,
                                                     image=item.image,
                                                     count=item.count,
                                                     price=item.price,
                                                     spec=[v.dict() for v in item.spec],
                                                     date=item.date,
                                                     time_duration=item.time_duration,
                                                     commit=False)

        # 插入日志记录
        flow_role_ids = [v.flow_role_id for v in flow.inform_config_obj if v.flow_process_id == flow_process_id]
        flow_role = await FlowRole.fetch_one(session, wheres=(FlowRole.id.in_(flow_role_ids), ~FlowRole.is_deleted))
        await FlowRecordLog.create_modify(session, flow_record_id=flow_record.id,
                                          flow_role_id=flow_role.id if flow_role else 0,
                                          flow_role_phone=flow_role.phone if flow_role else '',
                                          flow_process_id=flow_process_id,
                                          flow_process_name=flow_process_name, record_status=status, commit=False)

        await session.commit()

        # Step2: ==========发送短信、语音==========
        await RecordService.send_sms(session, flow_record_id=flow_record.id, sms_type='init')
        await RecordService.send_tel(session, flow_record_id=flow_record.id, tel_type='init')

        # Step3: ==========发送第三方平台==========
        await RecordService.post_third_party(session, flow_record_id=flow_record.id, device=device)

        # MQTT 推送
        await RecordService.push_mqtt(session, operate='add', device=device, flow_record=flow_record, response=response)

        return flow_record

    @staticmethod
    async def get_flow_record_detail(session, flow_record_id=0):
        """ 获取服务记录详情
        :param session:
        :param flow_record_id: 服务记录id
        :return:
        """
        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            return {}
        ret_data = flow_record.dict()

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        ret_data['flow_name'] = flow.name if flow else ''

        data = await FlowRecordDetail.fetch_all(session, wheres=(FlowRecordDetail.flow_record_id == flow_record.id), orders=(FlowRecordDetail.id, ))
        detail_map = {key: list(group) for key, group in groupby(data, key=lambda x: (x.node_id, x.node_type, x.node_name))}

        content = []
        for (node_id, node_type, node_name), detail in detail_map.items():
            # 图片列表、文本列表
            if node_type in ('image_display_select', 'text_picker'):
                items = [item.dict('title', 'content', 'image', 'price', 'count', 'detail', 'spec', 'group_title') for item in detail]
                content.append({'title': '服务内容', 'items': items, 'node_type': detail[0].node_type})

            elif node_type in ('service_time_booking',):
                result = {'title': '服务内容', 'node_type': 'service_time_booking'}
                for v in detail:
                    if v.time_duration:
                        result['time_duration'] = v.time_duration
                    if v.date:
                        result['date'] = v.date
                    if v.title:
                        result['value'] = v.title

                content.append(result)
            else:
                content.append({'title': detail[0].node_name, 'value': detail[0].title, 'date': detail[0].date, 'node_type': detail[0].node_type})

        ret_data['content'] = content

        # 流程状态信息
        flow_record_logs = await FlowRecordLog.fetch_all(session, wheres=(FlowRecordLog.flow_record_id == flow_record.id), orders=(FlowRecordLog.id.desc()))
        flow_role_ids = [v.flow_role_id for v in flow_record_logs]
        flow_roles = await FlowRole.fetch_all(session, wheres=(FlowRole.id.in_(flow_role_ids)))
        flow_role_map = {v.id: v for v in flow_roles}

        logs = []
        for flow_record_log in flow_record_logs:
            d = flow_record_log.dict()

            flow_role = flow_role_map.get(flow_record_log.flow_role_id)
            d['flow_role_name'] = flow_role.name if flow_role else ''
            logs.append(d)

        ret_data['flow_record_logs'] = logs

        # 发票信息
        invoice = await Invoice.fetch_one(session, wheres=(Invoice.flow_record_id == flow_record.id))
        ret_data['invoice'] = invoice.dict() if invoice else {}

        return ret_data

    @staticmethod
    async def do_flow_invoice_submit(session, body_params: BodyParams):
        """ 提交发票信息
        :param session:
        :param body_params:
        :return:
        """
        # 取得设备信息
        device: Device = await Zeus.get_device(body_params.dev_id)

        # 提交服务记录
        flow_record = await RecordService.do_flow_record_submit(session, flow_code=body_params.flow_code, device=device)
        if not flow_record:
            raise BizError('服务记录提交失败')

        # 绑定发票信息
        await Invoice.create_modify(session, flow_record_id=flow_record.id, **body_params.dict())

        return flow_record

    @staticmethod
    async def do_flow_resource_submit(session, body_params: BodyParams):
        """ 提交资源信息信息
        :param session:
        :param body_params:
        :return:
        """
        # 取得设备信息
        device: Device = await Zeus.get_device(body_params.dev_id)

        # 提交服务记录
        flow_record = await RecordService.do_flow_record_submit(session, flow_code=body_params.flow_code, device=device, resource=body_params.resource)
        if not flow_record:
            raise BizError('服务记录提交失败')

        return flow_record

    @staticmethod
    async def do_flow_record_cancel(session, code='', flow_code='', flow_record_id=0, reason='', dev_id=''):
        """ 取消服务记录
        :param session:
        :param code: 服务记录编号
        :param flow_code: 服务流程编号
        :param flow_record_id:
        :param dev_id: 设备号
        :param reason: 取消原因
        :return:
        """
        if code:
            flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.code == code))
        elif flow_code and dev_id:
            flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.dev_id == dev_id), orders=(FlowRecord.id.desc()))
            if flow_record and flow_record.flow_code != flow_code:
                return
        elif flow_record_id:
            flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        else:
            return

        if not flow_record or flow_record.status == 'cancel' or flow_record.status == 'success':
            return
        # 取消
        flow_record.status = status = 'cancel'
        flow_record.reason = reason

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        if not flow:
            raise BizError('服务流程不存在')

        # 插入记录
        flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow.id, ~FlowProcess.is_deleted, FlowProcess.rank == ProcessType.cancel.value.code))
        flow_record.flow_process_id = flow_process_id = flow_process.id if flow_process else 0
        flow_record.flow_process_name = flow_process_name = flow_process.name if flow_process else ProcessType.cancel.value.value
        response = flow_process.digital_human_broadcast if flow_process and flow_process.digital_human_broadcast else f'您的{flow.name}已取消，感谢您的使用'
        await FlowRecordLog.create_modify(session,
                                          flow_process_id=flow_process_id,
                                          flow_process_name=flow_process_name,
                                          flow_record_id=flow_record.id,
                                          flow_role_id=0, record_status=status, commit=False)

        await session.commit()

        # 发送短信、语音
        await RecordService.send_sms(session, flow_record_id=flow_record.id, sms_type='cancel')
        await RecordService.send_tel(session, flow_record_id=flow_record.id, tel_type='cancel')

        # MQTT 推送
        await RecordService.push_mqtt(session, operate='cancel', device=Device(dev_id=flow_record.dev_id, organization_code=flow_record.organization_code), flow_record=flow_record, response=response)

        return flow_record

    @staticmethod
    async def do_page_button_click(session, flow_record_id, flow_page_id, flow_role_id, submit_list):
        """ 流程跳转点击
        :param session:
        :param flow_record_id:
        :param flow_page_id:
        :param flow_role_id:
        :param submit_list:
        :return:
        """
        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            raise BizError('服务记录不存在')
        # 当前流程状态
        flow_process_id = flow_record.flow_process_id

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        if not flow:
            raise BizError('服务流程不存在')

        flow_page = await FlowPage.fetch_one(session, wheres=(FlowPage.id == flow_page_id, ~FlowPage.is_deleted))
        if not flow_page:
            raise BizError('页面组件不存在')

        # 目标流程状态
        flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.id == flow_page.to_flow_process_id, ~FlowProcess.is_deleted))
        if not flow_process:
            raise BizError('页面流程不存在')
        # 流转到该状态时，终端播报相应的内容
        response = flow_process.digital_human_broadcast

        sms_type = 'transfer'
        # 取消服务，直接取消，不需要校验流程流转
        if flow_process.rank == ProcessType.cancel.value.code:
            if flow_record.status == 'cancel':
                raise BizError('服务已取消')
            # 取消服务
            flow_record.status = sms_type = 'cancel'
            mqtt_operate = 'cancel'
            response = response if response else f'您的{flow.name}已取消，感谢您的使用'
        else:
            _flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow_record.flow_id, FlowProcess.rank < flow_process.rank, ~FlowProcess.is_deleted), orders=(FlowProcess.rank.desc()))
            if not _flow_process or _flow_process.id != flow_process_id:
                raise BizError('页面状态不支持流转')
            # 判断节点是否已完成
            _flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow_record.flow_id, FlowProcess.rank > flow_process.rank, ~FlowProcess.is_deleted))
            if not _flow_process:
                flow_record.status = sms_type = 'success'
            mqtt_operate = 'update'

        # 更新记录
        flow_record.flow_process_id = flow_process.id
        flow_record.flow_process_name = flow_process.name
        # 记录服务记录日志
        phone = ''
        flow_role = await FlowRole.fetch_one(session, wheres=(FlowRole.id == flow_role_id))
        if flow_role:
            # 用户
            if flow_role.role_type == 'user':
                flow_record_detail = await Select(selects=(FlowRecordDetail),
                                                  joins=((Node, FlowRecordDetail.node_id == Node.id),),
                                                  wheres=(FlowRecordDetail.flow_record_id == flow_record.id, Node.node_type == 'text_phone'),
                                                  orders=(FlowRecordDetail.id,)).fetch_one(session)

                if flow_record_detail and validate_phone(flow_record_detail.title):
                    phone = flow_record_detail.title
            # 工作人员
            elif flow_role.role_type == 'worker':
                phone = flow_role.phone

        await FlowRecordLog.create_modify(session,
                                          flow_record_id=flow_record.id,
                                          flow_role_id=flow_role_id,
                                          flow_role_phone=phone,
                                          flow_process_id=flow_process.id,
                                          flow_process_name=flow_process.name,
                                          record_status=flow_record.status,
                                          commit=False)
        # 保存组件的提交信息
        existing_data = flow_record.submit if flow_record.submit else []
        # 合并新的提交数据
        new_data = [v.dict() for v in submit_list]
        merged_data = existing_data + new_data
        flow_record.submit = json.dumps(merged_data)

        await session.commit()

        # MQTT 推送
        await RecordService.push_mqtt(session, operate=mqtt_operate, device=Device(dev_id=flow_record.dev_id, organization_code=flow_record.organization_code), flow_record=flow_record, response=response)

        # 发送完成短信
        await RecordService.send_sms(session, flow_record_id, sms_type=sms_type)

        return flow_record, flow_process

    @staticmethod
    async def do_page_submit_click(session, flow_record_id, submit_list):
        """ 页面提交点击
        :param session:
        :param flow_record_id: 服务记录id
        :param submit_list: 提交信息
        :return:
        """
        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            raise BizError('服务记录不存在')

        # 获取之前的提交信息
        existing_data = flow_record.submit if flow_record.submit else []
        # 合并新的提交数据
        new_data = [v.dict() for v in submit_list]
        merged_data = existing_data + new_data
        flow_record.submit = json.dumps(merged_data)

        await session.commit()

    @staticmethod
    async def do_flow_process_jump(session, flow_record_id, to_flow_process_id, role_name, role_phone, voice):
        """ 服务流程跳转
        :param session:
        :param flow_record_id:
        :param to_flow_process_id:
        :param role_name:
        :param role_phone:
        :param voice: 语音内容
        :return:
        """
        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.id == flow_record_id))
        if not flow_record:
            raise BizError('服务记录不存在')

        # 当前流程状态顺序
        current_flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.id == flow_record.flow_process_id, ~FlowProcess.is_deleted))
        if not current_flow_process:
            raise BizError('当前流程状态不存在')

        flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
        if not flow:
            raise BizError('服务流程不存在')

        # 目标流程状态
        flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.id == to_flow_process_id, ~FlowProcess.is_deleted))
        if not flow_process:
            raise BizError('目标流程状态不存在')
        # 流转到该状态时，终端播报相应的内容
        response = voice if voice else flow_process.digital_human_broadcast

        # 当前状态和目标状态是否一致
        if current_flow_process.id == flow_process.id:
            return

        jump_flow_process_list = []

        sms_type = 'transfer'
        # 取消服务，直接取消，不需要校验流程流转
        if flow_process.rank == ProcessType.cancel.value.code:
            if flow_record.status == 'cancel':
                raise BizError('服务已取消')
            # 取消服务
            flow_record.status = sms_type = 'cancel'
            mqtt_operate = 'cancel'
            response = response if response else f'您的{flow.name}已取消，感谢您的使用'
        else:
            jump_flow_process_list = await FlowProcess.fetch_all(session, wheres=(FlowProcess.flow_id == flow_record.flow_id, FlowProcess.rank < flow_process.rank, FlowProcess.rank > current_flow_process.rank, ~FlowProcess.is_deleted), orders=(FlowProcess.rank.asc()))
            # 判断节点是否已完成
            _flow_process = await FlowProcess.fetch_one(session, wheres=(FlowProcess.flow_id == flow_record.flow_id, FlowProcess.rank > flow_process.rank, ~FlowProcess.is_deleted))
            if not _flow_process:
                flow_record.status = sms_type = 'success'
            mqtt_operate = 'update'

        # 更新记录
        flow_record.flow_process_id = flow_process.id
        flow_record.flow_process_name = flow_process.name
        # 记录服务记录日志
        flow_role = await FlowRole.fetch_one(session, wheres=(FlowRole.flow_id == flow_record.flow_id, FlowRole.name == role_name, FlowRole.role_type == 'worker', ~FlowRole.is_deleted))
        if not role_phone and flow_role:
            role_phone = flow_role.phone

        if flow_record.status == 'cancel':
            await FlowRecordLog.create_modify(session,
                                              flow_record_id=flow_record.id,
                                              flow_role_id=flow_role.id if flow_role else 0,
                                              flow_role_phone=role_phone,
                                              flow_process_id=flow_process.id,
                                              flow_process_name=flow_process.name,
                                              record_status=flow_record.status,
                                              commit=False)
        else:
            for v in jump_flow_process_list:
                await FlowRecordLog.create_modify(session,
                                                  flow_record_id=flow_record.id,
                                                  flow_role_id=flow_role.id if flow_role else 0,
                                                  flow_role_phone=role_phone,
                                                  flow_process_id=v.id,
                                                  flow_process_name=v.name,
                                                  record_status='init',
                                                  commit=False)

            await FlowRecordLog.create_modify(session,
                                              flow_record_id=flow_record.id,
                                              flow_role_id=flow_role.id if flow_role else 0,
                                              flow_role_phone=role_phone,
                                              flow_process_id=flow_process.id,
                                              flow_process_name=flow_process.name,
                                              record_status=flow_record.status,
                                              commit=False)

        await session.commit()

        # MQTT 推送
        await RecordService.push_mqtt(session, operate=mqtt_operate, device=Device(dev_id=flow_record.dev_id, organization_code=flow_record.organization_code), flow_record=flow_record, response=response)
        # 发送完成短信
        await RecordService.send_sms(session, flow_record_id, sms_type=sms_type)

    @staticmethod
    async def get_flow_record_list(session, device, code=''):
        """ 查询指定设备的服务记录
        :param session: 数据库连接
        :param device: 设备信息
        :param code: 服务记录编码
        :return:
        """
        wheres = list()
        if code:
            wheres.append(FlowRecord.code == code)
        else:
            # 未处理完成以及在2分钟内处理完成的
            wheres.append(
                or_(
                    FlowRecord.status == 'init',
                    and_(
                        FlowRecord.status == 'success',
                        FlowRecord.update_time > datetime.now() - timedelta(minutes=2)
                    )
                )
            )
        wheres.append(FlowRecord.dev_id == device.dev_id)
        wheres.append(~FlowRecord.is_deleted)

        # 子查询(包含时间段节点)
        sub_query = Select(
            selects=(
                     FlowRecord.id.label('flow_record_id'),
                     FlowRecordDetail.date.label('date'),
                     FlowRecordDetail.title.label('title'),
                     case([
                         (FlowRecordDetail.node_type == 'time_period_picker',
                           func.STR_TO_DATE(
                               func.CONCAT(
                                   func.DATE_FORMAT(
                                       func.DATE_ADD(
                                           FlowRecord.create_time,
                                           text(
                                               "INTERVAL (CASE WHEN flow_record_detail.date = 'today' THEN 0 WHEN flow_record_detail.date = 'tomorrow' THEN 1 WHEN flow_record_detail.date = 'after_tomorrow' THEN 2 ELSE 0 END) DAY")),
                                       '%Y-%m-%d'
                                   ),
                                   func.SUBSTRING(FlowRecordDetail.title, 7, 5)
                               ),
                               '%Y-%m-%d %H:%i'
                           )),], else_=
                             func.STR_TO_DATE(
                               func.CONCAT(
                                   FlowRecordDetail.date,
                                   func.SUBSTRING(FlowRecordDetail.title, 0, 5)
                               ),
                               '%Y-%m-%d %H:%i'
                              )
                          ).label('appoint_time')
                     # func.STR_TO_DATE(
                     #     func.CONCAT(
                     #         func.DATE_FORMAT(
                     #             func.DATE_ADD(
                     #                 FlowRecord.create_time,
                     #                 text("INTERVAL (CASE WHEN flow_record_detail.date = 'today' THEN 0 WHEN flow_record_detail.date = 'tomorrow' THEN 1 WHEN flow_record_detail.date = 'after_tomorrow' THEN 2 ELSE 0 END) DAY")),
                     #             '%Y-%m-%d'
                     #         ),
                     #         func.SUBSTRING(FlowRecordDetail.title, 7, 5)
                     #     ),
                     #     '%Y-%m-%d %H:%i'
                     #     ).label('appoint_time')
                     ),
            joins=((FlowRecordDetail, FlowRecordDetail.flow_record_id == FlowRecord.id),),
            wheres=(
                FlowRecord.status == 'init',
                FlowRecord.dev_id == device.dev_id,
                ~FlowRecord.is_deleted,
                FlowRecordDetail.node_type.in_(['time_period_picker', 'service_time_booking'])
            )
        ).subquery()

        # 此处为了适用一个服务流程中有多个时间段选择页面, 增强系统鲁棒性
        sub_query_child = Select(
            selects=(sub_query.c.flow_record_id,
                     func.min(sub_query.c.appoint_time).label('appoint_time')),
            groups=(sub_query.c.flow_record_id, )
        ).subquery()

        sub_query = Select(
            selects=(
                sub_query.c.flow_record_id,
                sub_query.c.date,
                sub_query.c.title,
                sub_query.c.appoint_time
            ),
            joins=((sub_query_child, and_(
                sub_query_child.c.flow_record_id == sub_query.c.flow_record_id,
                sub_query_child.c.appoint_time == sub_query.c.appoint_time
            )),)
        ).subquery()

        # 过滤超时24小时以上的
        wheres.append(
            or_(
                sub_query.c.appoint_time.is_(None),
                func.TIMESTAMPDIFF(text('HOUR'), sub_query.c.appoint_time, func.now()) <= 24
            )
        )
        data = await FlowRecord.fetch_all(session,
                                          selects=(FlowRecord.id.label('flow_record_id'),
                                                   FlowRecord.code.label('code'),
                                                   Flow.id.label('flow_id'),
                                                   Flow.code.label('flow_code'),
                                                   Flow.name.label('flow_name'),
                                                   Flow.hint.label('flow_hint'),
                                                   FlowRecord.flow_process_name.label('flow_process_name'),
                                                   FlowRecord.status.label('status'),
                                                   FlowRecord.update_time.label('update_time'),
                                                   sub_query.c.date.label('date'),
                                                   sub_query.c.title.label('title'),
                                                   sub_query.c.appoint_time.label('appoint_time')),
                                          joins=((Flow, Flow.id == FlowRecord.flow_id), ),
                                          left_joins=((sub_query, sub_query.c.flow_record_id == FlowRecord.id),),
                                          # 有时间（超时 > 预计时间近 > 预计时间远）---没有时间
                                          orders=(
                                              FlowRecord.status,
                                              func.TIMESTAMPDIFF(text('SECOND'), sub_query.c.appoint_time, func.now()).desc()
                                          ),
                                          wheres=wheres)
        data = [dict(zip(v.keys(), v)) for v in data]
        flow_record_map = {v['flow_record_id']: v['flow_id'] for v in data}

        flow_record_ids = [v['flow_record_id'] for v in data]
        record_logs = await FlowRecordLog.fetch_all(session, wheres=(FlowRecordLog.flow_record_id.in_(flow_record_ids)), orders=(FlowRecordLog.id, ))
        # 已触达节点状态列表
        status_dict = {flow_record_id: [v.flow_process_name for v in _groupby if v.flow_process_name] for flow_record_id, _groupby in groupby(record_logs, key=lambda x: x.flow_record_id)}

        # 服务流程的工单状态列表信息
        flow_ids = [v['flow_id'] for v in data]
        processes = await FlowProcess.fetch_all(session, wheres=(FlowProcess.flow_id.in_(flow_ids),
                                                                 FlowProcess.rank >= 0,
                                                                 FlowProcess.is_display == 1,
                                                                 ~FlowProcess.is_deleted), orders=(FlowProcess.flow_id, FlowProcess.rank,))
        processes_map = {flow_id: [v.name for v in _groupby] for flow_id, _groupby in groupby(processes, key=lambda x: x.flow_id)}
        # 补齐未触达节点状态
        for flow_record_id, status_list in status_dict.items():
            flow_id = flow_record_map.get(flow_record_id, 0)
            processes = processes_map.get(flow_id)
            for process in processes or []:
                if process in status_list:
                    continue
                status_list.append(process)

        # 获取服务图标
        cariabot_flows = await FlowService.get_cariabot_flow_list(session, device)
        icon_dict = {v.code: v.image for v in cariabot_flows}

        for v in data:
            v['status_list'] = status_dict.get(v['flow_record_id'], [])
            if v['status_list']:
                v['flow_hint'] = ''
            date = v.pop('date')
            time = v.pop('title')
            if date and time:
                v['flow_time'] = {'date': date, 'time': time}
            v['icon'] = icon_dict.get(v['flow_code'], '')
            v['is_timeout'] = (v['appoint_time'] is not None and v['appoint_time'] < datetime.now())

        return data

    @staticmethod
    async def get_flow_record_detail_simplify(session, device: Device = None, code=''):
        """ 获取服务记录详情（精简版）
        :param session:
        :param device: 设备信息
        :param code: 服务记录编码
        :return:
        """
        flow_record = await FlowRecord.fetch_one(session, wheres=(FlowRecord.code == code, ~FlowRecord.is_deleted))
        if not flow_record:
            return {}
        ret_data = flow_record.dict('code', 'room_no', 'status', 'flow_process_name', 'update_time')
        # C端需要获取服务信息
        if device:
            # 获取服务图标
            cariabot_flows = await FlowService.get_cariabot_flow_list(session, device)
            icon_dict = {v.code: v.image for v in cariabot_flows}
            ret_data['icon'] = icon_dict.get(flow_record.flow_code, '')
            # 获取服务名称
            flow = await Flow.fetch_one(session, wheres=(Flow.id == flow_record.flow_id))
            ret_data['flow_name'] = flow.name if flow else ''
            ret_data['flow_hint'] = flow.hint if flow else ''

        # 获取服务日志
        flow_record_logs = await FlowRecordLog.fetch_all(session, wheres=(FlowRecordLog.flow_record_id == flow_record.id), orders=(FlowRecordLog.id,))
        status_list = [v.flow_process_name for v in flow_record_logs]

        # 获取服务工单状态
        processes = await FlowProcess.fetch_all(session, wheres=(FlowProcess.flow_id == flow_record.flow_id, FlowProcess.is_display, FlowProcess.rank >= 0, ~FlowProcess.is_deleted), orders=(FlowProcess.rank,))
        # 补齐未触达
        for process in processes:
            if process.name in status_list:
                continue
            status_list.append(process.name)

        ret_data['status_list'] = status_list
        # 获取服务详情
        details = await FlowRecordDetail.fetch_all(session, wheres=(FlowRecordDetail.flow_record_id == flow_record.id), orders=(FlowRecordDetail.id, ))
        content = []
        ret_data['is_timeout'] = False
        ret_data['appoint_time'] = None
        flow_time = {}
        for item in details:
            if item.node_type == 'image_display_select':    # 图片列表
                item_dict = item.dict('title', 'image', 'price', 'count', 'spec', 'node_type')
                item_dict['item_type'] = 'list_image'
                content.append(item_dict)
            elif item.node_type == 'text_picker':           # 文本列表
                item_dict = item.dict('title', 'image', 'price', 'count', 'spec', 'node_type')
                item_dict['item_type'] = 'list_text'
                content.append(item_dict)
            # 服务时间(康养)
            elif item.node_type == 'service_time_booking':
                if item.time_duration:
                    ret_data['time_duration'] = item.time_duration
                if item.date:
                    flow_time.update({'date': item.date})
                if item.title:
                    flow_time.update({'time': item.title})
                if flow_time:
                    ret_data['flow_time'] = flow_time
                ret_data['is_timeout'] = False
            elif item.node_type == 'time_period_picker':    # 时间选择器
                if item.date and item.title:
                    ret_data['flow_time'] = {'date': item.date, 'time': item.title}
                    days = 0
                    if item.date == 'today':
                        days = 0
                    elif item.date == 'tomorrow':
                        days = 1
                    elif item.date == 'after_tomorrow':
                        days = 2
                    ret_data['appoint_time'] = appoint_time = datetime.strptime(f'{(item.create_time + timedelta(days=days)).strftime("%Y-%m-%d")} {item.title[-5:]}', '%Y-%m-%d %H:%M')
                    ret_data['is_timeout'] = (appoint_time < datetime.now())
            else:
                content.append(item.dict('node_name', 'title', 'node_type'))
        ret_data['content'] = content

        return ret_data

    @staticmethod
    async def push_mqtt(session, operate, device: Device, flow_record: FlowRecord, response= ''):
        if not all([flow_record, flow_record.code, flow_record.flow_process_name]):
            logger.error(f'推送MQTT失败，flow_record为空或字段缺失')
            return

        data_list = await RecordService.get_flow_record_list(session, device=device, code=flow_record.code)
        data = serializer(data_list[0]) if data_list else {}
        push_data = {
            'dev_id': device.dev_id,
            'operate': operate,
            'data': data,
            'response': response
        }

        await MQTTService.push(topic=f'flow_record/{device.dev_id}', message=json.dumps(push_data, ensure_ascii=False).encode('utf-8'))


if __name__ == "__main__":
    import asyncio
    from project.lib.common import async_session
    db = async_session()
    result = asyncio.run(RecordService.get_flow_record_detail(db, flow_record_id=407))
    print(result)


